package cn.com.dhcc.ddi.task.service.extract;

import cn.com.dhcc.app.core.HookPara;
import cn.com.dhcc.app.core.annotation.SiColumn;
import cn.com.dhcc.app.core.dao.SimpleDaoForOracle;
import cn.com.dhcc.app.core.exception.DaoException;
import cn.com.dhcc.app.core.exception.NiceException;
import cn.com.dhcc.app.core.handler.dbhandler.DbTypeSynchronizationManagerMannager;
import cn.com.dhcc.app.core.handler.dbhandler.SingleTransationCircleWithOutResult;
import cn.com.dhcc.app.core.handler.modulehandler.SqlArg;
import cn.com.dhcc.app.core.service.Log;
import cn.com.dhcc.app.core.tag.CommonStaticCode;
import cn.com.dhcc.app.core.util.*;
import cn.com.dhcc.app.pub.core.consts.DdiProperties;
import cn.com.dhcc.app.pub.core.tag.staticcode.StaticCode;
import cn.com.dhcc.ddi.datasource.vo.DataSourceConfVo;
import cn.com.dhcc.ddi.datasource.vo.DataSourceVo;
import cn.com.dhcc.ddi.datasource.vo.TableColVo;
import cn.com.dhcc.ddi.datasource.vo.TableInfoVo;
import cn.com.dhcc.ddi.task.dao.TableLogDao;
import cn.com.dhcc.ddi.task.dao.TaskActiveLogDao;
import cn.com.dhcc.ddi.task.dao.TaskTabDao;
import cn.com.dhcc.ddi.task.service.ChangeVo;
import cn.com.dhcc.ddi.task.service.DbTypeConvertUtils;
import cn.com.dhcc.ddi.task.vo.TableLog;
import cn.com.dhcc.ddi.task.vo.Task;
import cn.com.dhcc.ddi.task.vo.TaskActiveLog;
import cn.com.dhcc.ddi.task.vo.TaskTab;
import cn.com.dhcc.app.pub.core.util.settlement.compute.*;
import org.apache.commons.io.FileUtils;
import org.apache.log4j.Logger;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.jdbc.core.ColumnMapRowMapper;
import org.springframework.jdbc.core.RowCallbackHandler;
import org.springframework.stereotype.Service;
import org.springframework.util.StringUtils;

import java.io.File;
import java.io.IOException;
import java.math.BigDecimal;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.text.DateFormat;
import java.text.ParseException;
import java.text.SimpleDateFormat;
import java.util.*;
import java.util.concurrent.ExecutionException;

/**
 * 抽取增量表数据
 * @日期：2016/11/15 15:27
  @作者：张洪涛
 * @版权所有：东华软件
 * @版本：1.0
 */
@Service
public class GatherChangeDbData {
    private static Logger logger = Log.getLogger(Log.LOGTYPE.DDI);
    @Autowired
    private TaskActiveLogDao taskActiveLogDao;
    @Autowired
    private TableLogDao tableLogDao;
    @Autowired
    private TaskTabDao taskTabDao;
    public static final int SETT_QUERY_NUM = DdiProperties.APP_SETT_MAXQUERYNUM;
    public static final int BATCH_QUERY_NUM = DdiProperties.APP_PAGE_MAXQUERYNUM;
    /**
     * 获取交换表数据
     * @param task
     * @param dataSource
     * @param tasksTables
     * @param tableName_tableLog
     * @param currentTimeSql
     * @param dataXmlPath
     * @return
     * @throws NiceException
     */
    public ChangeVo gatherChangeSql(final Task task, DataSourceVo dataSource, List<TableInfoVo> tasksTables, Map<String, TableLog> tableName_tableLog, final String taskLogId, String dataXmlPath) throws NiceException, DaoException {
        logger.info("进入方法gatherInsertChangeSql()。");
        final File f = new File(dataXmlPath);
        final ChangeVo changeVo = new ChangeVo();
        Map<String, List<Integer>> table_num = new LinkedHashMap<String, List<Integer>>();
        Map<String, String> startValues = new HashMap<String, String>();
        Map<String, String> endValues = new HashMap<String, String>();
        try {
            //是否为结算,判断一次性任务是否为临时新增的结算任务
            boolean isSett = false;
            if(task.getTaskFreqType().equals(StaticCode.TASK_FREQ_TYPE.ONETIME.getCode())&& task.getIsSett() !=null && "1".equals(task.getIsSett()) && task.getSettTaskId() != null){
                isSett = true;
            }
            boolean periodTaskSett = false;
            if(task.getTaskFreqType().equals(StaticCode.TASK_FREQ_TYPE.PERIOD.getCode())&& task.getIsSett() !=null && "1".equals(task.getIsSett())){
                periodTaskSett = true;
            }
            if (!checkTableDateCol(tasksTables)){
                periodTaskSett = false;
            }
            changeVo.setExecuteTableNames(new HashSet<String>());
            //初始化数据源
            final SimpleDaoForOracle daoForOracle = new SimpleDaoForOracle();
            DataSourceConfVo dsConfVo = JsonUtil.toObject(dataSource.getConf(),DataSourceConfVo.class);
            String driviceName = dsConfVo.getDriverName();
            String linkUrl = dsConfVo.getUrl();
            String username = dsConfVo.getUserName();
            String dsPasswd = dsConfVo.getPassword();
            final String dbtype = dsConfVo.getDbType();
            daoForOracle.initDataSource(driviceName, linkUrl, username, dsPasswd);

            if(!CollectionUtil.hasElement(tasksTables)){
                logger.error("没有要发送的列表。");
                return changeVo;
            }
            String currentTimeSql = StaticCode.DB_TYPE.find(dsConfVo.getDbType()).getValidateUrl();
            SimpleDateFormat sdf =   new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
            String dateString  = daoForOracle.queryForString("获取数据库系统时间", currentTimeSql, new Object[]{});
            Date currentTime = sdf.parse(dateString);
            for (final TableInfoVo table : tasksTables) {
                final List<Integer> sqlReturn = new ArrayList<Integer>();
                TableLog tableLog = null;
                String lastValue = null;
                String currentValue = null;
                TaskTab taskTab = taskTabDao.getTaskTabByTaskIdAndTableId(task.getId(), table.getId());
                String tableName = table.getSchema()+"."+ table.getTableName();
                if (tableName_tableLog != null) {
                    tableLog = tableName_tableLog.get(tableName);
                } else {
                    tableName_tableLog = new HashMap<String, TableLog>();
                }

                TableColVo icCol = getIC(table.getTableColVos());
                String startValue = taskTab.getStartValue();
                String endValue = taskTab.getEndValue();
                if(icCol != null){
                    if (tableLog != null) {//如果没有日志说明：是第一次同步,则全表同步，如果上次有时间，则从》上次时间的数据开始
                        if (tableLog.getLastValue() !=null && !tableLog.getLastValue().isEmpty()) {
                            lastValue = tableLog.getLastValue();
                            logger.info("获取到增量值为：" + lastValue);
                        } else if(taskTab.getStartValue() != null && !taskTab.getStartValue().isEmpty()){
                            logger.info("未获取到最后一次抽取增量值。");
                            lastValue = startValue;
                        }
                    } else {
                        logger.info("未获取到最后一次取增量值。");
                        if(taskTab.getStartValue() != null && !taskTab.getStartValue().isEmpty()){
                            lastValue = startValue;
                        }
                    }
                    SiColumn.DB_TYPE_GROUP group = SiColumn.DB_TYPE_GROUP.getTypeGroup(icCol.getColType());

                    if (SiColumn.DB_TYPE_GROUP.DATE == group || SiColumn.DB_TYPE_GROUP.TIMESTAMP == group) {
                        if(task.getTaskFreqType().equals(StaticCode.TASK_FREQ_TYPE.ONETIME.getCode())){
                            if(!taskTab.getEndValue().isEmpty()){
                                currentValue = taskTab.getEndValue();
                            }
                        }else{
                            currentValue = sdf.format(currentTime);
                        }

                    } else {
                        if(task.getTaskFreqType().equals(StaticCode.TASK_FREQ_TYPE.ONETIME.getCode())){
                            if(!taskTab.getEndValue().isEmpty()){
                                currentValue = taskTab.getEndValue();
                            }
                        }else{
                            String sql = "select max(" + icCol.getColName() + ") from " + tableName;
                            currentValue = daoForOracle.queryForString("获取最大增量值", sql, new Object[]{});
                        }
                    }
                    logger.info("抽取增量值为：" + lastValue + ";当前增量值：" + currentValue);
                }


                SqlArg sqlArg = createSelectDataSql(task, table, lastValue, currentValue,endValue);
                logger.debug("表[" + tableName + "]->args:" + Arrays.toString(sqlArg.getArgs()));
                changeVo.getExecuteTableNames().add(tableName);
                final List<Map<String, Object>> datas = new ArrayList<Map<String, Object>>();

                final TaskActiveLog log = taskActiveLogDao.getByTaskId(task.getId());
                if(isSett){
                    System.out.println("结算任务开始====>"+tableName);
                    final String toTaskId = task.getSettTaskId();
                    final Date settDate = task.getExtrStartTime();
                    //
                    Date start =  new Date();
                    List<String> pkStrValues = SettUtil.dataComputeResult(toTaskId, tableName, settDate);
                    if(pkStrValues == null || pkStrValues.size() == 0){
                        System.out.println("正在对比前一天交换数据");
                        daoForOracle.queryForRowSet("查询当天交换全部数据", sqlArg.getSql(),sqlArg.getArgs(),new RowCallbackHandler() {
                            ColumnMapRowMapper columnMapRowMapper = new ColumnMapRowMapper();
                            @Override
                            public void processRow(ResultSet rs) throws SQLException {
                                Map<String,Object> obj = columnMapRowMapper.mapRow(rs, 0);
                                datas.add(obj);
                                if(datas.size()== SETT_QUERY_NUM){
                                    try {
                                        int sum = createPkValuesData(datas, table,toTaskId,taskLogId,settDate);
                                        if(sum>0){
                                            log.setMtime(new Date());
                                            taskActiveLogDao.update(log);
                                        }
                                    } catch (NiceException e) {
                                        logger.error(e);
                                    }catch (IOException e) {
                                        logger.error(e);
                                    } catch (DaoException e) {
                                        logger.error(e);
                                    } catch (InterruptedException e) {
                                        logger.error(e);
                                    } catch (ExecutionException e) {
                                        logger.error(e);
                                    }
                                    datas.clear();
                                }
                            }
                        });
                        if(datas.size()>0){
                            int sum = createPkValuesData(datas, table,toTaskId,taskLogId,settDate);
                            if(sum>0){
                                //更新运行时间
                                log.setMtime(new Date());
                                taskActiveLogDao.update(log);
                            }
                            datas.clear();
                        }
                        System.out.println("对比结束");
                    }


                      //抽取数据
                    if(pkStrValues == null || pkStrValues.size() == 0){
                        pkStrValues = SettUtil.dataComputeResult(toTaskId, tableName, settDate);
                    }
                    final List<String> pks = pkStrValues;
                    if(pks != null && pks.size() > 0){
                        System.out.println("抽取结算数据开始");
                        datas.clear();
                        //在同一事务内抽取
                        daoForOracle.doInSingleTransationCircle("执行增量sql", new SingleTransationCircleWithOutResult() {
                            @Override
                            public void actionInCircle() throws RuntimeException {
                                try {
                                    SqlArg args = createSelectDataSqlForSett(table,pks.get(0),dbtype);
                                    List<TableColVo> cols = table.getTableColVos();
                                    Map<String,TableColVo> colMap = colsMap(cols);
                                    for (Iterator<String> iterator2 = pks.iterator(); iterator2.hasNext();) {
                                        String pkStr = iterator2.next();
                                        Object[] obj = createSelectArgs(colMap,pkStr,dbtype);
                                        daoForOracle.queryForRowSet("批量查询增量数据", args.getSql(),obj,new RowCallbackHandler() {
                                            ColumnMapRowMapper columnMapRowMapper = new ColumnMapRowMapper();
                                            @Override
                                            public void processRow(ResultSet rs) throws SQLException {
                                                Map<String,Object> obj = columnMapRowMapper.mapRow(rs, 0);
                                                datas.add(obj);
                                                if(datas.size()==BATCH_QUERY_NUM){
                                                    try {
                                                        ArrayList<String> batchSql = createXmlByData(datas, table,false,taskLogId);
                                                        if(CollectionUtil.hasElement(batchSql)){
                                                            //sqlReturn.size() 等于 0时，说明第一次抽取数据，数据文件写入表名
                                                            if(sqlReturn.size() == 0 && batchSql.size()>0){
                                                                FileUtils.writeStringToFile(f, "<table name='"+ table.getTableName()+"'>", "utf-8",true);
                                                            }
                                                            if(batchSql.size()>0){
                                                                sqlReturn.add(batchSql.size());
                                                            }
                                                            FileUtils.writeLines(f, "utf-8", batchSql,true);
                                                            //更新运行时间
                                                            log.setMtime(new Date());
                                                            DbTypeSynchronizationManagerMannager.bindDbType(HookPara.DB_FEATURE);
                                                            taskActiveLogDao.update(log);
                                                            changeVo.setTableDataFlag(true);
                                                            batchSql.clear();
                                                        }
                                                    } catch (NiceException e) {
                                                        logger.error(e);
                                                    }catch (IOException e) {
                                                        logger.error(e);
                                                    } catch (DaoException e) {
                                                        logger.error(e);
                                                    } catch (ParseException e) {
                                                        logger.error(e);
                                                    }
                                                    datas.clear();
                                                }
                                            }
                                        });

                                    }
                                    if(datas.size()>0){
                                        ArrayList<String> batchSql = createXmlByData(datas, table,false,taskLogId);
                                        if(CollectionUtil.hasElement(batchSql)){
                                            //sqlReturn.size() 等于 0时，说明第一次抽取数据，数据文件写入表名
                                            if(sqlReturn.size() == 0 && batchSql.size()>0){
                                                FileUtils.writeStringToFile(f, "<table name='"+ table.getTableName()+"'>", "utf-8",true);
                                            }
                                            if(batchSql.size()>0){
                                                sqlReturn.add(batchSql.size());
                                            }
                                            FileUtils.writeLines(f, "utf-8", batchSql,true);
                                            changeVo.setTableDataFlag(true);
                                            batchSql.clear();
                                        }
                                        datas.clear();
                                    }
                                } catch (NiceException e) {
                                    logger.error(e);
                                }catch (IOException e) {
                                    logger.error(e);
                                } catch (DaoException e) {
                                    logger.error(e);
                                } catch (ParseException e) {
                                    logger.error(e);
                                }
                            }
                        });
                        System.out.println("抽取结算数据结束");
                    }
                    Date end=  new Date();
                    long interval = (end.getTime() - start.getTime())/1000;
                    System.out.println("-------------用时："+interval);
                    System.out.println("结算任务结束");
                }else{
                    final boolean needSett = periodTaskSett;
                    daoForOracle.queryForRowSet("批量查询增量数据", sqlArg.getSql(),sqlArg.getArgs(),new RowCallbackHandler() {
                        ColumnMapRowMapper columnMapRowMapper = new ColumnMapRowMapper();
                        @Override
                        public void processRow(ResultSet rs) throws SQLException {
                            Map<String,Object> obj = columnMapRowMapper.mapRow(rs, 0);
                            datas.add(obj);
                            if(datas.size()== BATCH_QUERY_NUM){
                                try {
                                    ArrayList<String> batchSql = createXmlByData(datas, table,needSett,taskLogId);
                                    if(CollectionUtil.hasElement(batchSql)){
                                        //sqlReturn.size() 等于 0时，说明第一次抽取数据，数据文件写入表名
                                        if(sqlReturn.size() == 0 && batchSql.size()>0){
                                            FileUtils.writeStringToFile(f, "<table name='"+ table.getTableName()+"'>", "utf-8",true);
                                        }
                                        if(batchSql.size()>0){
                                            sqlReturn.add(batchSql.size());
                                        }

                                        FileUtils.writeLines(f, "utf-8", batchSql,true);
                                        //更新运行时间
                                        log.setMtime(new Date());
                                        taskActiveLogDao.update(log);
                                        changeVo.setTableDataFlag(true);
                                        batchSql.clear();
                                    }
                                } catch (NiceException e) {
                                    logger.error(e);
                                }catch (IOException e) {
                                    logger.error(e);
                                } catch (DaoException e) {
                                    logger.error(e);
                                } catch (ParseException e) {
                                    logger.error(e);
                                }
                                datas.clear();
                            }
                        }
                    });
                    if(datas.size()>0){
                        ArrayList<String> batchSql = createXmlByData(datas, table,needSett,taskLogId);
                        if(CollectionUtil.hasElement(batchSql)){
                            //sqlReturn.size() 等于 0时，说明第一次抽取数据，数据文件写入表名
                            if(sqlReturn.size() == 0 && batchSql.size()>0){
                                FileUtils.writeStringToFile(f, "<table name='"+ table.getTableName()+"'>", "utf-8",true);
                            }
                            if(batchSql.size()>0){
                                sqlReturn.add(batchSql.size());
                            }
                            FileUtils.writeLines(f, "utf-8", batchSql,true);
                            changeVo.setTableDataFlag(true);
                            batchSql.clear();
                        }
                    }
                }
//                daoForOracle.queryForRowSet("批量查询增量数据", sqlArg.getSql(), sqlArg.getArgs(), new RowCallbackHandler() {
//                    ColumnMapRowMapper columnMapRowMapper = new ColumnMapRowMapper();
//                    @Override
//                    public void processRow(ResultSet rs) throws SQLException {
//                        Map<String, Object> obj = columnMapRowMapper.mapRow(rs, 0);
//                        datas.add(obj);
//                        if (datas.size() == DdiProperties.APP_PAGE_MAXQUERYNUM) {
//                            try {
//                                ArrayList<String> batchSql = createXmlByData(datas, tab,false,taskLogId);
//                                if (CollectionUtil.hasElement(batchSql)) {
//                                    //sqlReturn.size() 等于 0时，说明第一次抽取数据，数据文件写入表名
//                                    if (sqlReturn.size() == 0 && batchSql.size() > 0) {
//                                        FileUtils.writeStringToFile(f, "<table name='" + tab.getTableName() + "'>", "utf-8", true);
//                                    }
//                                    if (batchSql.size() > 0) {
//                                        sqlReturn.add(batchSql.size());
//                                    }
//                                    FileUtils.writeLines(f, "utf-8", batchSql, true);
//                                    //更新运行时间
//                                    log.setMtime(new Date());
//                                    taskActiveLogDao.update(log);
//                                    changeVo.setTableDataFlag(true);
//                                    batchSql.clear();
//                                }
//                            } catch (NiceException e) {
//                                logger.error(e);
//                            } catch (IOException e) {
//                                logger.error(e);
//                            } catch (DaoException e) {
//                                logger.error(e);
//                            } catch (ParseException e) {
//                                logger.error(e);
//                            }
//                            datas.clear();
//                        }
//                    }
//                });
//                if (datas.size() > 0) {
//                    ArrayList<String> batchSql = createXmlByData(datas, tab,false,taskLogId);
//                    if (CollectionUtil.hasElement(batchSql)) {
//                        //sqlReturn.size() 等于 0时，说明第一次抽取数据，数据文件写入表名
//                        if (sqlReturn.size() == 0 && batchSql.size() > 0) {
//                            FileUtils.writeStringToFile(f, "<table name='" + tab.getTableName() + "'>", "utf-8", true);
//                        }
//                        if (batchSql.size() > 0) {
//                            sqlReturn.add(batchSql.size());
//                        }
//                        FileUtils.writeLines(f, "utf-8", batchSql, true);
//                        changeVo.setTableDataFlag(true);
//                        batchSql.clear();
//                    }
//                }
                //记录表日志
                if (tableLog == null) {
                    tableLog = new TableLog();
                    tableLog.setId(UUIDGenerator.getUUID());
                    tableLog.setTaskId(task.getId());
                    tableLog.setTableName(tableName);
                    tableName_tableLog.put(tableName, tableLog);
                }
                //加入同步sql
                if (changeVo.isTableDataFlag()) {
                    if(null != icCol){
                        tableLog.setTempLastValue(currentValue);
                    }
                    tableLog.setMtime(new Date());
                    if (sqlReturn.size() > 0) {
                        int count = 0;
                        for (Integer sql : sqlReturn) {
                            count += sql;
                        }
                        if (count > 0) {
                            table_num.put(tableName, sqlReturn);
                            startValues.put(tableName, lastValue);
                            endValues.put(tableName, currentValue);
                        }
                        FileUtils.writeStringToFile(f, "</table>", "utf-8", true);
                    }
                }
                tableLogDao.insertOrUpdate(tableLog);
            }
            changeVo.setTableDataNum(table_num);
            changeVo.setStartValue(startValues);
            changeVo.setEndValue(endValues);
        } catch (IOException e) {
            logger.error(e);
        } catch (ParseException e) {
            logger.error(e);
        } catch (DaoException e) {
            logger.error("增量数据检索异常",e);
            throw new DaoException("增量数据检索异常!",e);
        } catch (InterruptedException e) {
            throw new NiceException("增量数据检索异常!",e);
        } catch (ExecutionException e) {
            throw new NiceException("增量数据检索异常!",e);
        }
        logger.info("执行结束gatherInsertChangeSql()。");
        return changeVo;
    }

    /**
     * 生成增量数据
     * @param taskTab
     * @param lastValue
     * @param currentValue
     * @return
     * @throws NiceException
     */
    private SqlArg createSelectDataSql(Task task,TableInfoVo taskTab,String lastValue,String currentValue,String endValue) throws NiceException, ParseException {
        logger.info("进入方法createSelectDataSql()。");
        SqlArg sqlArg = new SqlArg();
        StringBuilder selectSql = new StringBuilder("select  ");
        List<TableColVo> cols = taskTab.getTableColVos();
        if(!CollectionUtil.hasElement(cols)){
            throw new NiceException("表["+taskTab.getTableName()+"]未发布字段!");
        }
        for (Iterator<TableColVo> iterator = taskTab.getTableColVos().iterator(); iterator.hasNext();) {
            TableColVo taskTabCol = iterator.next();
            selectSql.append(taskTabCol.getColName());
            if(iterator.hasNext())selectSql.append(",");
        }
        TableColVo icCol = getIC(taskTab.getTableColVos());
        selectSql.append(" from ").append(taskTab.getSchema()).append(".").append(taskTab.getTableName());
        if(StringUtils.hasText(taskTab.getWhereCondition())){
            if(!taskTab.getWhereCondition().trim().toLowerCase().startsWith("where")){
                selectSql.append(" where ");
            }
            selectSql.append(" ").append(taskTab.getWhereCondition());
        }else{
            if(null != icCol) {
                selectSql.append(" where 1=1 ");
            }
        }


        //如果是周期性任务，增量字段做为数据抽取条件，是否使用增量字段
        if(null != icCol){
            selectSql.append(" and ");
            //根据字段类型转换增量值类型
            Object lastVal = formatValue(icCol.getColName(),icCol.getColType(), lastValue);
            Object currentVal = formatValue(icCol.getColName(),icCol.getColType(), currentValue);
            Object endVal = formatValue(icCol.getColName(),icCol.getColType(), endValue);
            if ( endVal == null){
                endVal = currentVal;
            }

            SiColumn.DB_TYPE_GROUP group = SiColumn.DB_TYPE_GROUP.getTypeGroup(icCol.getColType());
            //时间类型增量值向前播n秒容错
            if (lastVal != null){
                if(SiColumn.DB_TYPE_GROUP.DATE == group || SiColumn.DB_TYPE_GROUP.TIMESTAMP == group){
                    SimpleDateFormat dateFormat= new SimpleDateFormat("yyyy-MM-dd hh:mm:ss");
                    Date lastTime = (Date)lastVal;
                    Calendar cal = Calendar.getInstance();
                    cal.setTime(lastTime);
                    cal.add(Calendar.SECOND,-DdiProperties.APP_CORRECTIONTIME);
                    lastTime = cal.getTime();
                    lastVal = lastTime;
                }
                selectSql.append(icCol.getColName()).append(" > ? and ").append(icCol.getColName()).append(" <= ? and ").append(icCol.getColName()).append(" <= ? ");
                sqlArg.setArgs(new Object[]{lastVal,currentVal,endVal});
            }else{
                selectSql.append(icCol.getColName()).append(" <= ? and ").append(icCol.getColName()).append(" <= ? ");
                sqlArg.setArgs(new Object[]{currentVal,endVal});
            }
        }else{

        }
        logger.info("[createSelectDataSql]Return sql:" + selectSql.toString());
        sqlArg.setSql(selectSql.toString());
        return sqlArg;
    }
    /**
     * 根据字段类型格式化值
     * @param type
     * @param value
     * @return
     */
    private Object formatValue(String name, String type, String value){
        if (value==null || "".equals(value)) {//如果value的值为null 或者为""
            return null;
        }
        try {
            return DbTypeConvertUtils.convert(type, value);
        } catch (ParseException e) {
            logger.error("字段["+name+"]转换异常，type["+type+"],name["+name+"]", e);
        }
        return value;
    }
    private ArrayList<String> createXmlByData(
            List<Map<String, Object>> datasList,TableInfoVo taskTab,boolean needSett,String taskLogId) throws NiceException, ParseException {
        logger.info("进入方法createXmlByData()。");
        ArrayList<String> arrayList = new ArrayList<String>();
        List<String> pkValues = new ArrayList<String>();
        List<TableColVo>  pkCols = getPkCols(taskTab);
        for (Map<String, Object> row : datasList) {
            String rowData = createXmlBySingleRow(row, taskTab);
            if (StrUtil.isNotBank(rowData)) {
                arrayList.add("<row>" + rowData + "</row>");
            }
            String pkValuesStr = getPkValuesString(pkCols, row);
            //拼接时间字段
            TableColVo icCol = getIC(taskTab.getTableColVos());
            //如果时间戳字段不作为发送字段则不对时间戳字段hash
            if(icCol != null && icCol.getColName() != null && icCol.getColType() != null){
                String ntColName = icCol.getColName();
                Object ntValue = row.get(ntColName);
                pkValuesStr += "|"+ntColName+"=>"+getColDataToString(icCol, ntValue);
            }
            pkValues.add(pkValuesStr);
        }

        if(needSett && pkValues.size() > 0){
            String tableName = taskTab.getSchema()+"."+taskTab.getTableName();
            SettUtil.dataStore(tableName,pkValues,taskLogId);
        }
        logger.info("执行结束createXmlByData()。");
        return arrayList;
    }

    private String createXmlBySingleRow(Map<String, Object> row,
                                        TableInfoVo taskTab) throws NiceException, ParseException {
        logger.info("进入方法createInsertXmlBySingleRow()。");
        StringBuilder insertXml = new StringBuilder();
        List<TableColVo> cols = taskTab.getTableColVos();
        //特殊情况，如果任务配置了修改和删除，但是此时修改了上次同步记录的时间戳字段为大于上次同步时间，此时抽取出insert语句，就会出现违反唯一性约束的问题。
        //解决办法为：在此条件下，不抽取此条pk对应记录的insert语句，让update去做
        for (TableColVo taskTabCol : cols) {
            insertXml.append("<col colName='").append(taskTabCol.getColName()).append("' colType='").append(taskTabCol.getColType()).append("' isPK='").append(taskTabCol.getIsPk()).append("'>");
            String value = getTypeConvertWarpCol(taskTabCol, row.get(taskTabCol.getColName()));
//            if(taskTabCol.getColType())
            insertXml.append("<![CDATA[").append(getTruncateValue(taskTabCol,value)).append("]]>");
            insertXml.append("</col>");
        }
        logger.info("[createInsertXmlBySingleRow]Return xml:" + insertXml.toString());
        return insertXml.toString();
    }

    /**
     * 获取增字段
     * @param cols
     * @return
     */
    private TableColVo getIC(List<TableColVo> cols){
        for (TableColVo tabCol : cols) {
            if (tabCol.getIsIc().equals(CommonStaticCode.YESNO.YES.getStatus())){
                return tabCol;
            }
        }
        return null;
    }

    /**
     * 封装字段值
     * @param tabCol
     * @param data
     * @return
     * @throws NiceException
     * @throws ParseException
     */
    private String getTypeConvertWarpCol(TableColVo tabCol,Object data) throws NiceException, ParseException {
        String colType = tabCol.getColType();
        SiColumn.DB_TYPE_GROUP group = SiColumn.DB_TYPE_GROUP.getTypeGroup(colType);
        if(data==null || "".equals(data+"")){
            return "";
        }
        if(SiColumn.DB_TYPE_GROUP.STR == group){
            return ""+data;
        }
        else if(SiColumn.DB_TYPE_GROUP.NUMBER == group){
            String dataStr = data+"";
            BigDecimal bigDecimal = new BigDecimal(dataStr.trim());
            return bigDecimal.toPlainString()+"";
        }
        else if(SiColumn.DB_TYPE_GROUP.DATE == group){
            DateFormat dateFormat = new SimpleDateFormat(DateUtil.FMT_yyyymmddHHmmss);
            String javadateStr;
            javadateStr = dateFormat.format((Date)data);
            return javadateStr;

        }
        else if(SiColumn.DB_TYPE_GROUP.TIMESTAMP == group){
            DateFormat dateFormat = new SimpleDateFormat(DateUtil.FMT_yyyymmddhh24mmssSSS);
            String javadateStr;
            javadateStr = dateFormat.format((Date)data);
            return javadateStr;
        }else{
            throw new NiceException("不支持的数据库字段类型["+colType+"]!");
        }
    }

    /**
     * 截取带有.0的整数  如  111.0  截取为111
     * @aram tabCol
     * @param data
     * @return
     */
    private String getTruncateValue(TableColVo tabCol, String data){
        String colType = tabCol.getColType();
        SiColumn.DB_TYPE_GROUP group = SiColumn.DB_TYPE_GROUP.getTypeGroup(colType);
        if(SiColumn.DB_TYPE_GROUP.NUMBER == group){
            if (data!= null && !"".equals(data) && (Math.round(Math.floor(Double.parseDouble(data)))  == Double.parseDouble(data)) && data.indexOf(".0") != -1 ){
                data =  Math.round(Math.floor(Double.parseDouble(data)))+"";
            }
        }
        return data;
    }

    /**
     *拼接主键和增量字段
     * @param datasList
     * @param tab
     * @param taskId
     * @param taskLogId
     * @param date
     * @return
     * @throws NiceException
     * @throws IOException
     * @throws InterruptedException
     * @throws ExecutionException
     */
    private int createPkValuesData(
            List<Map<String, Object>> datasList,TableInfoVo tab,String taskId,String taskLogId,Date date) throws NiceException, IOException, InterruptedException, ExecutionException {
        logger.info("进入方法createPkValuesData()。");
        List<String> pkValues = new ArrayList<String>();
        List<TableColVo>  pkCols = getPkCols(tab);
        for (Iterator<Map<String, Object>> iterator = datasList.iterator(); iterator.hasNext();) {
            Map<String, Object> row =  iterator.next();
            //以前同步过，但是时间错字段被修改了。这个时候不抽取insert，而是抽取update
            String pkValuesStr = getPkValuesString(pkCols, row);
            //拼接增量字段
            TableColVo icCol = getIC(tab.getTableColVos());
            if(icCol != null && icCol.getColName() != null && icCol.getColType() != null){
                String ntColName = icCol.getColName();
                Object ntValue = row.get(ntColName);
                pkValuesStr += "|"+ntColName+"=>"+getColDataToString(icCol, ntValue);
            }

            pkValues.add(pkValuesStr);
        }
        String tableName = tab.getSchema()+"."+tab.getTableName();
        int sum = SettUtil.dataCompute(taskId, tableName, pkValues, taskLogId,date);

        logger.info("执行结束createPkValuesData()。");
        return sum;
    }
    /**
     * 根据发布的表，获取到表的主键或者唯一建字段
     * @param strucSTaskTab
     * @return
     */
    public List<TableColVo> getPkCols(TableInfoVo tab){
        List<TableColVo> pkCols = new ArrayList<TableColVo>();
        List<TableColVo> cols = tab.getTableColVos();
        for (Iterator<TableColVo> iterator = cols.iterator(); iterator.hasNext();) {
            TableColVo tabCol = iterator.next();
            if(CommonStaticCode.YESNO.YES.getStatus().equalsIgnoreCase(tabCol.getIsPk())){
                pkCols.add(tabCol);
            }
        }
        return pkCols;
    }

    /**
     * 获取主键值拼接支付串，作为key——hash的key.such as-->  id:asdfafasdsfsdfs|name:qiaobing
     * @param pkCols
     * @param row
     * @return
     * @throws NiceException
     */
    public String getPkValuesString(List<TableColVo> pkCols,Map<String, Object> row) throws NiceException{
        String pkValuesStr = "";
        for (Iterator<TableColVo> iterator = pkCols.iterator(); iterator.hasNext();) {
            TableColVo pkCol = iterator.next();
            if(!pkCol.getIsPk().equalsIgnoreCase(CommonStaticCode.YESNO.YES.getStatus()))continue;
            String pkColName = pkCol.getColName();
            Object pkValue = row.get(pkColName);
            pkValuesStr += pkColName+"=>"+getColDataToString(pkCol, pkValue);
            if(iterator.hasNext())
                pkValuesStr+="|";
        }
        return pkValuesStr;
    }

    /**
     * 根据字段类型获取当前字段值的toString值
     * @param tabCol
     * @param data
     * @return
     * @throws NiceException
     */
    public String getColDataToString(TableColVo tabCol,Object data) throws NiceException{
        String colType = tabCol.getColType();
        SiColumn.DB_TYPE_GROUP group = SiColumn.DB_TYPE_GROUP.getTypeGroup(colType);
        if(SiColumn.DB_TYPE_GROUP.STR == group){
            return data+"";
        }
        else if(SiColumn.DB_TYPE_GROUP.NUMBER == group){
            BigDecimal bigDecimal = new BigDecimal(data.toString().trim());
            return bigDecimal.doubleValue()+"";
        }
        else if(SiColumn.DB_TYPE_GROUP.DATE == group){
            DateFormat dateFormat = new SimpleDateFormat(DateUtil.FMT_yyyymmddHHmmss);
            String javadateStr;
            javadateStr = dateFormat.format((Date)data);
            return javadateStr;
        }
        else if(SiColumn.DB_TYPE_GROUP.TIMESTAMP == group){
            DateFormat dateFormat = new SimpleDateFormat(DateUtil.FMT_yyyymmddhh24mmssSSS);
            String javadateStr;
            javadateStr = dateFormat.format((Date)data);
            return javadateStr;
        }else{
            throw new NiceException("不支持的数据库字段类型["+colType+"]!");
        }
    }

    private SqlArg createSelectDataSqlForSett(TableInfoVo tab,String pkStr,String dbtype) throws NiceException {
        logger.info("进入方法createSelectDataSql()。");
        SqlArg sqlArg = new SqlArg();
        StringBuffer selectInsertSql = new StringBuffer("select  ");
        List<TableColVo> cols = tab.getTableColVos();
        Map<String,TableColVo> colMap = colsMap(cols);
        if(!CollectionUtil.hasElement(cols)){
            throw new NiceException("表["+tab.getTableName()+"]未发布字段!");
        }
        for (Iterator<TableColVo> iterator = cols.iterator(); iterator.hasNext();) {
            TableColVo col = iterator.next();
            selectInsertSql.append(col.getColName());
            if(iterator.hasNext())selectInsertSql.append(",");
        }
        selectInsertSql.append(" from ").append(tab.getSchema()+"."+tab.getTableName());
        if(StringUtils.hasText(tab.getWhereCondition())){
            if(!tab.getWhereCondition().trim().toLowerCase().startsWith("where")){
                selectInsertSql.append(" where ");
            }
            selectInsertSql.append(" ").append(tab.getWhereCondition());
        }else{
            selectInsertSql.append(" where 1=1 ");
        }
        //拼接sql，主键做为查询条件,去掉时间戳字段
        String pks[] = pkStr.split("\\|");

        Object[] args = new Object[pks.length];
        for(int i = 0;i<pks.length;i++){
            String pk = pks[i];
            String colVal[] = pk.split("=>");
            String col = colVal[0];
            String val = colVal[1];
            TableColVo colvo = colMap.get(col.toUpperCase());
            if (SiColumn.DB_TYPE_GROUP.getTypeGroup(colvo.getColType()).equals(SiColumn.DB_TYPE_GROUP.DATE) || SiColumn.DB_TYPE_GROUP.getTypeGroup(colvo.getColType()).equals(SiColumn.DB_TYPE_GROUP.TIMESTAMP)){
                if(dbtype.equals(StaticCode.DB_TYPE.SQLSERVER.getCode())|| dbtype.equals(StaticCode.DB_TYPE.SQLSERVEREXT.getCode())){
                    selectInsertSql.append(" and ").append("convert(varchar,"+col+",120)").append("=?").append("");
                    continue;
                }
            }
            selectInsertSql.append(" and ").append(col).append("=?").append("");

        }
        logger.info("[createSelectDataSql]Return sql:" + selectInsertSql.toString());
        sqlArg.setSql(selectInsertSql.toString());
        return sqlArg;
    }

    private Object[] createSelectArgs(Map<String,TableColVo> colsMap,String pkStr,String dbType ) throws NiceException {
        //拼接sql，主键做为查询条件,去掉时间戳字段
        String pks[] = pkStr.split("\\|");
        Object[] args = new Object[pks.length];
        for(int i = 0;i<pks.length;i++){
            String pk = pks[i];
            String colVal[] = pk.split("=>");
            String colString = colVal[0];
            String valString = colVal[1];
            TableColVo col = colsMap.get(colString.toUpperCase());
            Object val = null;
            if (SiColumn.DB_TYPE_GROUP.getTypeGroup(col.getColType()).equals(SiColumn.DB_TYPE_GROUP.DATE) || SiColumn.DB_TYPE_GROUP.getTypeGroup(col.getColType()).equals(SiColumn.DB_TYPE_GROUP.TIMESTAMP)){
                if(dbType.equals(StaticCode.DB_TYPE.SQLSERVER.getCode())|| dbType.equals(StaticCode.DB_TYPE.SQLSERVEREXT.getCode())){
                    val = valString;
                }else{
                    val = formatValue(col.getColName(),col.getColType(), valString);
                }
            }else{
                val = formatValue(col.getColName(),col.getColType(), valString);
            }
            args[i] = val;
        }

        return args;
    }

    public Map<String,TableColVo> colsMap(List<TableColVo> cols){
        Map<String,TableColVo> colsMap=new HashMap<String,TableColVo>();//表的对应信息
        for (Iterator<TableColVo> iterator = cols.iterator(); iterator.hasNext();) {
            TableColVo col = iterator.next();
            colsMap.put(col.getColName().toUpperCase(), col);
        }
        return colsMap;
    }

    /**
     * 判断任务所有表增量字段是否为时间字段
     * @param tasksTables
     * @return
     */
    private boolean checkTableDateCol(List<TableInfoVo> tasksTables){
        boolean flag = false;
        int timeic = 0;
        for (TableInfoVo tab:tasksTables) {
            TableColVo icCol = getIC(tab.getTableColVos());
            if(icCol != null){
                SiColumn.DB_TYPE_GROUP group = SiColumn.DB_TYPE_GROUP.getTypeGroup(icCol.getColType());
                if (SiColumn.DB_TYPE_GROUP.DATE == group || SiColumn.DB_TYPE_GROUP.TIMESTAMP == group) {
                    timeic++;
                }
            }

        }
        if (timeic == tasksTables.size()){
            flag = true;
        }
        return flag;
    }
}
